package com.dmp.report

import com.dmp.beans.Logs
import com.dmp.utils.ReportUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 需求：3.2.1 地域分布（省份城市报表开发）
  * 运行参数：G:\光环国际大数据开发班\大数据最后阶段-项目\21-dmp项目\资料\data.txt xx xxx
  */
object ProvinceCityReport {

  def main(args: Array[String]): Unit = {
    if(args.length < 3){
      println(
        """
          |com.dmp.report.ProvinceCityReport <logInputPath> <provniceDataPath> <cityDataPath>
          |<logInputPath> 文件输入目录
          |<provniceDataPath> 省份结果文件目录
          | <cityDataPath> 城市结果文件目录
        """.stripMargin)
      System.exit(0)
    }
    val Array(loginputpath,provincedatapath,citydatapath)=args

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(s"${this.getClass.getSimpleName}")
    //使用Kryo序列化
    conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    conf.registerKryoClasses(Array(classOf[Logs]))
    val sc = new SparkContext(conf)

    val provniceCityRDD: RDD[(String, String, List[Double])] = sc.textFile(loginputpath).map(line => {
      val log = Logs.line2Log(line)
      //统计请求数
      val adRequest = ReportUtils.calculateRequest(log)
      //计算竞价数
      val adResponse = ReportUtils.calculateResponse(log)
      //计算展示量和点击量
      val adClick = ReportUtils.calculateShowClick(log)
      //用于计算广告消费和广告成本
      val adCost = ReportUtils.calculateAdCost(log)
      //返回 设备所在省份名称，设备所在城市名称，请求数+竞价数+展示量和点击量+广告消费和广告成本
      (log.provincename, log.cityname, adRequest ++ adResponse ++ adClick ++ adCost)
      //cache()的作用：变换前后的新旧RDD的分区在物理上可能是同一块内存存储，这是Spark内部做的优化。有些RDD是计算的中间结果，其分区并不一定有相对应的内存或磁盘数据与之对应，所以如果想要复用某一个RDD，需要通过Cache算子，将数据缓存（或者说固化）到内存中。
    }).cache()

    /**
      * 省份的结果
      */
    provniceCityRDD.map( tuple =>{
      (tuple._1,tuple._3)
    }).reduceByKey{
      case(list1, list2) =>{
        //List[(Int, Int)] = List((1,5), (2,6), (3,7), (4,8))
        list1.zip(list2).map{
          //List[(Int, Int)] = List(6, 8, 10, 12)
          case (x,y) => x + y
        }
      }
        //最后返回的是(各省份,15)这种的
    }.foreach( tuple =>{
      val provinceName = tuple._1
      val report = tuple._2.mkString(",")
      println(provinceName + " "+ report)
    })

    /**
      * 城市的结果
      */

    provniceCityRDD.map( tuple =>{
      (tuple._1 + tuple._2,tuple._3)
    }).reduceByKey{
      case(list1, list2) =>{
        list1.zip(list2).map{
          case (x,y) => x + y
        }
      }
    }.foreach( tuple =>{
      val provniceAndCityName = tuple._1
      val report = tuple._2.mkString(",")
      println(provniceAndCityName + " "+ report)
    })

    sc.stop()
  }

}
